package com.tn.mqtt;

import com.tn.utils.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @ClassName Callback
 * @Description TODO
 * @Author Yezhongzhu
 * @Date 2021/11/26 9:43
 * @Version v1.0
 **/
@Slf4j
public class Callback implements MqttCallback {
    @Autowired
    private MQTTConnect server;

    @Autowired
    private ThreadPoolExecutor executor;

    /**
     * MQTT 断开连接会执行此方法
     */
    @Override
    public void connectionLost(Throwable throwable) {
        log.info("断开了MQTT连接 ：{}", throwable.getMessage());
        log.error(throwable.getMessage(), throwable);
    }

    /**
     * publish发布成功后会执行到这里
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        log.info("发布消息成功");
    }

    /**
     * subscribe订阅后得到的消息会执行到这里
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws MqttException{
        //  TODO    此处可以将订阅得到的消息进行业务处理、数据存储
        executor.execute(()->{
            try {
                consumer(topic,new String(message.getPayload(), StandardCharsets.UTF_8));
            } catch (MqttException e) {
                log.error("业务异常："+e);
            }
        });


    }

    public void consumer(String topic,String msg) throws MqttException {
        log.info("收到来自 " + topic + " 的消息：{}", msg);
    }

    public  void init(String topicc,Callback callback) throws Exception{
        server.setMqttClient(server.mqttConfig.getUserName(),server.mqttConfig.getPassWord(),callback);
        server.sub(topicc);
    }
}
